package com.study.rabbitmq.workQueue;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.study.rabbitmq.utils.RabbitMqUtil;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * 发布确认消息
 * 1.单个确认、
 * 2.批量确认
 * 3.异步批量确认
 */
public class ConfirmMessage {
    private static final Integer MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        //  1.单个确认
//        singletonConfirm(); //203 ms
        //  2.批量确认
//        batchConfirm();  //53 ms
        //  3.异步批量确认  单例
//        asynchConfirm();    //19 ms
    }

    /**
     * 单个确认
     *
     * @throws IOException
     * @throws InterruptedException
     * @throws TimeoutException
     */
    public static void singletonConfirm() throws IOException, InterruptedException, TimeoutException {
        Channel channel = RabbitMqUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        //开启消息发布确认
        channel.confirmSelect();
        //记录当前系统时间
        Long startTiem = System.currentTimeMillis();
        for (Integer i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            //发送是否成功
            boolean confirm = channel.waitForConfirms();
            if (confirm) {
                System.out.println(message + "消息发送成功");
            }
        }
        Long endTime = System.currentTimeMillis();
        String format = String.format("使用单个确认发布%d条消息所消耗的时间为：%d ms", MESSAGE_COUNT, endTime - startTiem);
        System.out.println(format);
        channel.close();
    }

    //批量确认消息的个数
    private static Integer batchCount = 100;

    public static void batchConfirm() throws IOException, InterruptedException {
        //1.创建信道
        Channel channel = RabbitMqUtil.getChannel();
        //2.创建队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        //3.开启信道确认
        channel.confirmSelect();
        Long startTime = System.currentTimeMillis();
        for (Integer i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            if (i % batchCount == 0) {
                boolean confirm = channel.waitForConfirms();
                if (confirm)
                    System.out.printf("发送第%d个%d个消息成功\n", i / batchCount + 1, batchCount);
                else
                    System.out.printf("发送第%d个%d个消息失败\n", i / batchCount + 1, batchCount);
            }
        }
        Long endTime = System.currentTimeMillis();
        System.out.printf("所有消息使用批量确认发送完毕耗时：%d ms", endTime - startTime);
    }

    public static void asynchConfirm() throws IOException {
        //1.创建信道
        Channel channel = RabbitMqUtil.getChannel();
        //2.创建队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        //3.开启信道确认
        channel.confirmSelect();
        Long startTime = System.currentTimeMillis();
        /**
         * 创建一个线程安全的有序的hash表，适用于高并发场景；
         * 1.存储序号和消息
         * 2.可以根据序号获取消息、
         * 3.支持并发访问
         */
        ConcurrentSkipListMap<Long, String> outstandConfirm = new ConcurrentSkipListMap<>();

        //信道确认调用的回调函数   deliveryTag为每条消息的键值
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple){
                //删除集合中已经确认的消息
                ConcurrentNavigableMap<Long, String> confirmed = outstandConfirm.headMap(deliveryTag);
                confirmed.clear();
            } else {
                outstandConfirm.remove(deliveryTag);
            }
            System.out.println("发送成功的消息为：" + deliveryTag);
        };
        //信道未确认调用的回调方法
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            System.out.println("发送失败的消息为：" + outstandConfirm.get(deliveryTag));
        };
        //4.增加信道确认监听
        channel.addConfirmListener(ackCallback, nackCallback);
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            //5.发送消息
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            outstandConfirm.put(channel.getNextPublishSeqNo(),message);
        }
        Long endTime = System.currentTimeMillis();
        System.out.printf("使用异步确认方式发送消息耗时为：%d ms", endTime - startTime);
    }

}
